package com.atguigu.chapter07.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/20 11:40
 */
public class Flink01_State_Operate_List {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.enableCheckpointing(1000); // 每s做一次Checkpoint
        
        env
            .socketTextStream("hadoop162", 9999)  // 在socket终端只输入毫秒级别的时间戳
            .flatMap(new MyFlatMapFunction())
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static class MyFlatMapFunction implements FlatMapFunction<String, String>, CheckpointedFunction {
        
        private ArrayList<String> words = new ArrayList<>();
        private ListState<String> wordList;
    
        @Override
        public void flatMap(String line, Collector<String> out) throws Exception {
            for (String word : line.split(" ")) {
                words.add(word);
            }
            
            out.collect(words.toString());
            
            
            if (line.contains("a")) {
                throw new RuntimeException();
            }
            
        }
        
        // 把状态进行快照(保存状态)
        @Override
        public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
            // 周期性执行:
            //            System.out.println("MyFlatMapFunction.snapshotState");
            // 获取算子状态中的列表状态
//            wordList.addAll(words); // 追加
            wordList.update(words); //覆盖
            
            
        }
        
        // 初始化状态: 读取状态
        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            // 程序启动的执行: 有几个并行度, 就执行几次
            System.out.println("MyFlatMapFunction.initializeState");
            // flink会从持久化存储的地方把数据恢复
            wordList = ctx.getOperatorStateStore().getListState(new ListStateDescriptor<String>("wordList", String.class));
    
            Iterable<String> it = wordList.get();
            for (String word : it) {
                words.add(word);
            }
    
        }
    }
}
